package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/24 11:39
 */
public class Dwd_16_DwdUserRegisterApp extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_16_DwdUserRegisterApp().init(2016, 2, "Dwd_16_DwdUserRegisterApp");
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        
        readOdsDb(tEnv, "Dwd_16_DwdUserRegisterApp");
    
        Table useInfo = tEnv.sqlQuery("select " +
                                        "data['id'] user_id, " +
                                        "data['create_time'] create_time, " +
                                        "date_format(data['create_time'], 'yyyy-MM-dd') date_id, " +
                                        "ts " +
                                        "from ods_db " +
                                        "where `table`='user_info' " +
                                        "and `type`='insert'");
    
        tEnv.executeSql("create table `dwd_user_register`( " +
                                "`user_id` string, " +
                                "`date_id` string, " +
                                "`create_time` string, " +
                                "`ts` string " +
                                ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_USER_REGISTER));
    
        useInfo.executeInsert("dwd_user_register");
    
    
    
    }
}
